package com.bj58.sparksql.hive.main;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.hive.HiveContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class TestHive {
	 private static Logger log = LoggerFactory.getLogger(TestHive.class);
	 private static HiveContext sqlContext = null;
	 static{
			//new SparkConf().setAppName(appName).setMaster(master);
			SparkConf conf = new SparkConf()
					.set("spark.default.parallelism", "30")	//reduceByKeyAndWindow执行时启动的线程数，默认是8个
					.set("spark.yarn.driver.memoryOverhead", "1024")
					.set("spark.yarn.executor.memoryOverhead", "2048")
					.set("spark.storage.memoryFraction", "0.5");
			JavaSparkContext sc = new JavaSparkContext(conf);
			sqlContext = new org.apache.spark.sql.hive.HiveContext(sc.sc());
			sqlContext.sql("use data");
	 }
	 
	 private static class SQLTask implements Callable<Integer>{
		 private String querySQL = null;
		 private long taskID = 0;
		 
		public SQLTask(String querySQL, long taskID) {
			this.querySQL = querySQL;
			this.taskID = taskID;
		}

		@Override
		public Integer call() throws Exception {
			log.info("====begin query hive===="+taskID);
			DataFrame df = sqlContext.sql(querySQL);
			log.error(df.first().mkString());
			df.show();
			log.info("=====end query hive====="+taskID);
			return 0;
		}
		 
	 }
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		String querySQL = "select  count(1) from data.t_jingzhun_user_stat_new where stat_date='2015-11-01' and business_type_biz='zhaopin'";
		log.warn("=========begin=========");
		ExecutorService threadPool = Executors.newCachedThreadPool();
		threadPool.submit(new SQLTask(querySQL, 1));
		querySQL = "select sum(cost)/100 ,sum(cash)/100 ,sum(gift)/100 ,sum(coupon)/100,count(distinct if(cost>0,user_id,null)),sum(cost)/count(distinct if(cost>0,user_id,null))/100 ,sum(cost)/sum(click)/100,count(distinct if(disp>0,user_id,null)),count(distinct if(cost>0,user_id,null)) from data.t_jingzhun_user_stat_new where stat_date='2015-11-01' and business_type_biz='zhaopin'";
		threadPool.submit(new SQLTask(querySQL, 2));
		threadPool.submit(new SQLTask(querySQL,3));
		threadPool.submit(new SQLTask(querySQL, 4));
		try {
			threadPool.shutdown();
	        // (所有的任务都结束的时候，返回TRUE)
	        if(!threadPool.awaitTermination(5, TimeUnit.MINUTES)){
	            // 超时的时候向线程池中所有的线程发出中断(interrupted)。
	        	threadPool.shutdownNow();
	        }
		} catch (InterruptedException e) {
			threadPool.shutdownNow();
			e.printStackTrace();
		}
		log.warn("=========end=========");
		
	}

}
